package com.lingyun.mqttserver.config;

import jakarta.annotation.PreDestroy;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Configuration
public class MqttConfig {
    // MQTT服务器地址
    private static final String MQTT_BROKER_URL = "tcp://47.93.160.65:1883";
    // 基础客户端ID，实际使用时会添加唯一标识
    private static final String BASE_CLIENT_ID = "springboot-mqtt-client234";
    // 用户名
    private static final String USERNAME = "admin";
    // 密码
    private static final String PASSWORD = "6868lingyun.";

    // 订阅的主题 - 可以根据不同业务类型定义不同主题
    private static final String[] TOPICS = {"test/#", "demo/topic", "device/+/status", "+/+/report", "+/+/connect", "+/submit"};

    // 订阅的QoS级别
    private static final int[] QOS = {1, 1, 1, 1, 1, 1};

    // 创建线程池 - 新增
    private final ExecutorService mqttMessageExecutor = Executors.newFixedThreadPool(10);

    /**
     * 配置MQTT客户端工厂
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();

        options.setServerURIs(new String[]{MQTT_BROKER_URL});
        options.setUserName(USERNAME);
        options.setPassword(PASSWORD.toCharArray());
        // 断开连接后是否自动重连
        options.setAutomaticReconnect(true);
        // 是否清除会话
        options.setCleanSession(true);
        // 设置连接超时时间
        options.setConnectionTimeout(30);
        // 设置保持连接心跳时间
        options.setKeepAliveInterval(60);

        factory.setConnectionOptions(options);
        return factory;
    }

    /**
     * 配置MQTT入站消息适配器（订阅消息）
     * 使用UUID确保clientId唯一
     */
    @Bean
    public MqttPahoMessageDrivenChannelAdapter inbound() {
        // 添加UUID确保每次连接的clientId唯一
        String clientId = BASE_CLIENT_ID + "_" + UUID.randomUUID().toString().substring(0, 8);

        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(clientId,
                        mqttClientFactory(),
                        TOPICS);
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(QOS);

        // 可以添加消息路由器，将不同主题的消息路由到不同的通道
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    /**
     * 消息路由 - 将特定主题的消息路由到专用通道
     */
    @Bean
    public MessageHandler messageRouter() {
        return message -> {
            String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
            if (topic.startsWith("device/")) {
                System.out.println("我到了哦~");
                deviceStatusChannel().send(message);
            } else {
                mqttInputChannel().send(message);
            }
        };
    }

    /**
     * 接收消息的通道 - 通用通道
     */
    @Bean
    public MessageChannel mqttInputChannel() {
//        return new DirectChannel();
        return new ExecutorChannel(mqttMessageExecutor);
    }

    /**
     * 接收设备状态消息的专用通道
     */
    @Bean
    public MessageChannel deviceStatusChannel() {
        return new DirectChannel();
    }

    /**
     * 设备状态消息处理器 - 专门处理设备状态消息
     */
    @Bean
    @ServiceActivator(inputChannel = "deviceStatusChannel")
    public MessageHandler deviceStatusHandler() {
        System.out.println("我到了");
        return message -> {
            String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
            String payload = message.getPayload().toString();
            System.out.println("[设备状态处理器] 收到消息 - 主题: " + topic + ", 内容: " + payload);
            // 这里可以添加设备状态处理的业务逻辑
        };
    }



    /**
     * 发送消息的通道
     */
    @Bean
    public MessageChannel mqttOutputChannel() {
        return new DirectChannel();
    }

    /**
     * 配置MQTT出站消息处理器（发送消息）
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutputChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
                BASE_CLIENT_ID + "_sender_" + UUID.randomUUID().toString().substring(0, 8),
                mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(1);
        messageHandler.setDefaultTopic("default/topic");
        messageHandler.setConverter(new DefaultPahoMessageConverter());
        return messageHandler;
    }
}
